package com.jinzheng.nettytestss.commont.kafkaconfig;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.config.KafkaListenerContainerFactory;
import org.springframework.kafka.core.*;
import org.springframework.kafka.listener.ConcurrentMessageListenerContainer;
import org.springframework.kafka.listener.ContainerProperties;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

@Configuration
@ConditionalOnProperty(prefix = "KafkaTemplateConfig", name = "isOpen", havingValue = "true")
public class KafkaTemplateConfig {

    @Autowired
    private KafkaBaseProp kafkaBaseProp;
    @Autowired
    private KafkaConsumerProp kafkaConsumerProp;
    @Autowired
    private KafkaProducerProp kafkaProducerProp;
    @Autowired
    private KafkaListenerProp kafkaListenerProp;

    /**
     * Producer Template 配置
     */
    @Bean
    public KafkaTemplate<String, String> kafkaTemplate() {
        return new KafkaTemplate<>(producerFactory());
    }

    /**
     * Producer 工厂配置
     */
    @Bean
    public ProducerFactory<String, String> producerFactory() {
        return new DefaultKafkaProducerFactory<>(producerConfigs());
    }

    /**
     * Producer 参数配置
     */
    @Bean
    public Map<String, Object> producerConfigs() {
        Map<String, Object> props = new ConcurrentHashMap<>();
        //kafka地址
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaBaseProp.getBootstrapServers());
        //保证幂等性、消息顺序性
        props.put(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, kafkaProducerProp.getMaxInFlightRequestsPerConnection());
        //只能保证单分区上的幂等性
        props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, kafkaProducerProp.getEnableIdempotence());
        /**
         *acks = 0 如果设置为零，则生产者将不会等待来自服务器的任何确认，该记录将立即添加到套接字缓冲区并视为已发送。在这种情况下，无法保证服务器已收到记录，并且重试配置将不会生效（因为客户端通常不会知道任何故障），为每条记录返回的偏移量始终设置为 - 1。
         *acks = 1 这意味着leader会将记录写入其本地日志，但无需等待所有副本服务器的完全确认即可做出回应，在这种情况下，如果leader在确认记录后立即失败，但在将数据复制到所有的副本服务器之前，则记录将会丢失。
         *acks = all 这意味着leader将等待完整的同步副本集以确认记录，这保证了只要至少一个同步副本服务器仍然存活，记录就不会丢失，这是最强有力的保证，这相当于acks = -1 的设置。
         */
        props.put(ProducerConfig.ACKS_CONFIG, kafkaProducerProp.getAcks());
        //去缓冲区中一次拉16k的数据，发送到broker
        props.put(ProducerConfig.BATCH_SIZE_CONFIG, kafkaProducerProp.getBatchSize());
        //设置缓存区大小
        props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, kafkaProducerProp.getBufferMemory());
        // 生产者空间不足时，send()被阻塞的时间，默认60s
        props.put(ProducerConfig.MAX_BLOCK_MS_CONFIG, kafkaProducerProp.getMaxBlockMs());
        // 批量发送，延迟为30毫秒，如果30ms内凑不够batch则强制发送，提高并发
        props.put(ProducerConfig.LINGER_MS_CONFIG, kafkaProducerProp.getLingerMs());
        // 消息的最大大小限制,也就是说send的消息大小不能超过这个限制, 默认1048576(1MB)
        props.put(ProducerConfig.MAX_REQUEST_SIZE_CONFIG, kafkaProducerProp.getMaxRequestSize());
        // 压缩消息，支持四种类型，分别为：none、lz4、gzip、snappy，默认为none。
        // 消费者默认支持解压，所以压缩设置在生产者，消费者无需设置。
        props.put(ProducerConfig.COMPRESSION_TYPE_CONFIG,kafkaProducerProp.getCompressionType());
        //消息发送失败重试次数（默认是Integer.MAX_VALUE）
        props.put(ProducerConfig.RETRIES_CONFIG,kafkaProducerProp.getRetries());
        //重试间隔时间（ms）
        props.put(ProducerConfig.RETRY_BACKOFF_MS_CONFIG,kafkaProducerProp.getRetryBackoffMs());
        //key、value的序列化方式
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,kafkaProducerProp.getKeySerializer());
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, kafkaProducerProp.getValueSerializer());
        return props;
    }

    /**
     * 监听工厂
     * */
    @Bean
    KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<Object, Object>> kafkaListenerContainerFactory() {
        ConcurrentKafkaListenerContainerFactory<Object, Object> factory = new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory());
        //线程数
        factory.setConcurrency(kafkaListenerProp.getConcurrency());
        //手动提交
        factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL);
        //开启批量处理
        factory.setBatchListener(kafkaListenerProp.getBatchListener());
        factory.getContainerProperties().setPollTimeout(kafkaListenerProp.getPollTimeout());
        return factory;
    }

    /**
     * kafka消费者工厂
     * */
    @Bean
    public ConsumerFactory<Object, Object> consumerFactory() {
        return new DefaultKafkaConsumerFactory(consumerConfigs());
    }

    @Bean
    public Map<String,Object> consumerConfigs() {
        Map<String,Object> props = new ConcurrentHashMap<>();
        //配置地址
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaBaseProp.getBootstrapServers());
        //key、value的反序列化方式
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, kafkaConsumerProp.getKeyDeserializer());
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, kafkaConsumerProp.getValueDeserializer());
        //消费者组
        props.put(ConsumerConfig.GROUP_ID_CONFIG, kafkaConsumerProp.getGroupId());
        //是否开启自动提交
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, kafkaConsumerProp.getEnableAutoCommit());
        //消费策略
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, kafkaConsumerProp.getAutoOffsetReset());
        //消费者默认等待服务响应时间(毫秒)
        props.put(ConsumerConfig.FETCH_MAX_WAIT_MS_CONFIG, kafkaConsumerProp.getFetchMaxWait());
        return props;
    }


}